package com.lvy.hczn.mq.rocket;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.lvy.hczn.front.common.constant.RedisConstants;
import com.lvy.hczn.front.common.core.redis.RedisCache;
import com.lvy.hczn.front.common.exception.UtilException;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Configuration
@EnableConfigurationProperties(MqProperties.class)
@ConditionalOnProperty(prefix = "rocketmq.lvy", value = "nameSrvAddr")
@Slf4j
public class MqConfiguration {

    @Autowired
    private MqProperties mqProperties;

    @Autowired
    private RedisCache redisCache;

    /**
     * 初始化向rocketmq发送普通消息的生产者
     */
    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.lvy", value = "producerInstanceName")
    public DefaultMQProducer defaultProducer() throws MQClientException {
        /**
         * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
         * 注意：ProducerGroupName需要由应用来保证唯一<br>
         * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
         * 因为服务器会回查这个Group下的任意一个Producer
         */
        String suffix = warehouse();
        DefaultMQProducer producer = new DefaultMQProducer(mqProperties.getProducerGroupName() + suffix, getAclRPCHook());
//        DefaultMQProducer producer = new DefaultMQProducer(mqProperties.getProducerGroupName());
        producer.setNamesrvAddr(mqProperties.getNameSrvAddr());
        producer.setInstanceName(mqProperties.getProducerInstanceName() + suffix);
        if (!StrUtil.isEmpty(mqProperties.getUnitName())) {
            producer.setUnitName(mqProperties.getUnitName() + suffix);
        }
        producer.setVipChannelEnabled(false);
        producer.setRetryTimesWhenSendAsyncFailed(10);
        /**
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        producer.start();
        return producer;
    }

    /**
     * 初始化向rocketmq发送事务消息的生产者
     */
    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.lvy", value = "producerTranInstanceName")
    public TransactionMQProducer transactionProducer() {
        /**
         * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
         * 注意：ProducerGroupName需要由应用来保证唯一<br>
         * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
         * 因为服务器会回查这个Group下的任意一个Producer
         */
        String suffix = warehouse();
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer(mqProperties.getTransactionProducerGroupName() + suffix, getAclRPCHook());
//        TransactionMQProducer producer = new TransactionMQProducer(mqProperties.getTransactionProducerGroupName());
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-tbs-thread");
                return thread;
            }
        });
        producer.setExecutorService(executorService);
        producer.setTransactionListener(transactionListener);
        producer.setNamesrvAddr(mqProperties.getNameSrvAddr());
        producer.setInstanceName(mqProperties.getProducerTranInstanceName() + suffix);
        producer.setRetryTimesWhenSendAsyncFailed(10);
        /**
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        try {
            producer.start();
        } catch (MQClientException e) {
            log.error("mq事物生产者初始化异常");
            e.printStackTrace();
        }
        return producer;
    }

    private RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(mqProperties.getAclAccessKey(), mqProperties.getAclSecretKey()));
    }

    private String warehouse() {
        Object warehouse = redisCache.get(RedisConstants.WAREHOUSE_WMS);
        if (warehouse == null) {
            return "";
        }
        Map<String, Object> map = BeanUtil.beanToMap(warehouse);
        if (map == null || map.isEmpty()) {
            throw new UtilException("仓库信息为空");
        }
        if (StrUtil.isEmpty(String.valueOf(map.get("areaCode")))) {
            throw new UtilException("仓库运营区域为空");
        }
        String suffix = "_" + String.valueOf(map.get("warehouseCode"));
        return suffix;
    }
}
